transitions with django models?In this comment proofit404 provided a nice example about how to use transitions and django together:
In [ ]:
from django.db import models
from django.db.models.signals import post_init
from django.dispatch import receiver
from django.utils.translation import ugettext_lazy as _
from transitions import Machine
class ModelWithState(models.Model):
ASLEEP = 'asleep'
HANGING_OUT = 'hanging out'
HUNGRY = 'hungry'
SWEATY = 'sweaty'
SAVING_THE_WORLD = 'saving the world'
STATE_TYPES = [
(ASLEEP, _('asleep')),
(HANGING_OUT, _('hanging out')),
(HUNGRY, _('hungry')),
(SWEATY, _('sweaty')),
(SAVING_THE_WORLD, _('saving the world')),
]
state = models.CharField(
_('state'),
max_length=100,
choices=STATE_TYPES,
default=ASLEEP,
help_text=_('actual state'),
)
@receiver(post_init, sender=ModelWithState)
def init_state_machine(instance, **kwargs):
states = [state for state, _ in instance.STATE_TYPES]
machine = instance.machine = Machine(model=instance, states=states, initial=instance.state)
machine.add_transition('work_out', instance.HANGING_OUT, instance.HUNGRY)
machine.add_transition('eat', instance.HUNGRY, instance.HANGING_OUT)
transitions memory footprint is too large for my Django app and adding models takes too long.We analyzed the memory footprint of transitions in this discussion and could verify that the standard approach is not suitable to handle thousands of models. However, with a static (class) machine and some __getattribute__ tweaking we can keep the convenience loss minimal:
In [1]:
from transitions import Machine
from functools import partial
from mock import MagicMock
class Model(object):
machine = Machine(model=None, states=['A', 'B', 'C'], initial=None,
transitions=[
{'trigger': 'go', 'source': 'A', 'dest': 'B', 'before': 'before'},
{'trigger': 'check', 'source': 'B', 'dest': 'C', 'conditions': 'is_large'},
], finalize_event='finalize')
def __init__(self):
self.state = 'A'
self.before = MagicMock()
self.after = MagicMock()
self.finalize = MagicMock()
@staticmethod
def is_large(value=0):
return value > 9000
def __getattribute__(self, item):
try:
return super(Model, self).__getattribute__(item)
except AttributeError:
if item in self.machine.events:
return partial(self.machine.events[item].trigger, self)
raise
model = Model()
model.go()
assert model.state == 'B'
assert model.before.called
assert model.finalize.called
model.check()
assert model.state == 'B'
model.check(value=500)
assert model.state == 'B'
model.check(value=9001)
assert model.state == 'C'
assert model.finalize.call_count == 4
Currently, transitions has no such callback. This example from the issue discussed here might give you a basic idea about how to extend Machine with such a feature:
In [4]:
from transitions.core import Machine, State, Event, EventData, listify
class DuringState(State):
# add `on_during` to the dynamic callback methods
# this way on_during_<state> can be recognized by `Machine`
dynamic_methods = State.dynamic_methods + ['on_during']
# parse 'during' and remove the keyword before passing the rest along to state
def __init__(self, *args, **kwargs):
during = kwargs.pop('during', [])
self.on_during = listify(during)
super(DuringState, self).__init__(*args, **kwargs)
def during(self, event_data):
for handle in self.on_during:
event_data.machine.callback(handle, event_data)
class DuringEvent(Event):
def _trigger(self, model, *args, **kwargs):
# a successful transition returns `res=True` if res is False, we know that
# no transition has been executed
res = super(DuringEvent, self)._trigger(model, *args, **kwargs)
if res is False:
state = self.machine.get_state(model.state)
event_data = EventData(state, self, self.machine, model, args=args, kwargs=kwargs)
event_data.result = res
state.during(event_data)
return res
class DuringMachine(Machine):
# we need to override the state and event classes used by `Machine`
state_cls = DuringState
event_cls = DuringEvent
class Model:
def on_during_A(self):
print("Dynamically assigned callback")
def another_callback(self):
print("Explicitly assigned callback")
model = Model()
machine = DuringMachine(model=model, states=[{'name': 'A', 'during': 'another_callback'}, 'B'],
transitions=[['go', 'B', 'A']], initial='A', ignore_invalid_triggers=True)
machine.add_transition('test', source='A', dest='A', conditions=lambda: False)
assert not model.go()
assert not model.test()
This has been a feature request here. We'd encourage to write a wrapper which converts a condensed statement into individual condition-based transitions. However, a less expressive version could look like this:
In [11]:
from transitions import Machine, Transition
from six import string_types
class DependingTransition(Transition):
def __init__(self, source, dest, conditions=None, unless=None, before=None,
after=None, prepare=None, **kwargs):
self._result = self._dest = None
super(DependingTransition, self).__init__(source, dest, conditions, unless, before, after, prepare)
if isinstance(dest, dict):
try:
self._func = kwargs.pop('depends_on')
except KeyError:
raise AttributeError("A multi-destination transition requires a 'depends_on'")
else:
# use base version in case transition does not need special handling
self.execute = super(DependingTransition, self).execute
def execute(self, event_data):
func = getattr(event_data.model, self._func) if isinstance(self._func, string_types) \
else self._func
self._result = func(*event_data.args, **event_data.kwargs)
super(DependingTransition, self).execute(event_data)
@property
def dest(self):
return self._dest[self._result] if self._result is not None else self._dest
@dest.setter
def dest(self, value):
self._dest = value
# subclass Machine to use DependingTransition instead of standard Transition
class DependingMachine(Machine):
transition_cls = DependingTransition
def func(value):
return value
m = DependingMachine(states=['A', 'B', 'C', 'D'], initial='A')
# define a dynamic transition with a 'depends_on' function which will return the required value
m.add_transition(trigger='shuffle', source='A', dest=({1: 'B', 2: 'C', 3: 'D'}), depends_on=func)
m.shuffle(value=2) # func returns 2 which makes the transition dest to be 'C'
assert m.is_C()
Note that this solution has some drawbacks. For instance, the generated graph might not include all possible outcomes.
Machine.get_triggers should only show valid transitions based on some conditions.This has been requested here. Machine.get_triggers is usually quite naive and only checks for theoretically possible transitions. If you need more sophisticated peeking, this PeekMachine._can_trigger might be a solution:
In [24]:
from transitions import Machine, EventData
from functools import partial
class Model(object):
def fails(self, condition=False):
return False
def success(self, condition=False):
return True
# condition is passed by EventData
def depends_on(self, condition=False):
return condition
def is_state_B(self, condition=False):
return self.state == 'B'
class PeekMachine(Machine):
def _can_trigger(self, model, *args, **kwargs):
# We can omit the first two arguments state and event since they are only needed for
# actual state transitions. We do have to pass the machine (self) and the model as well as
# args and kwargs meant for the callbacks.
e = EventData(None, None, self, model, args, kwargs)
return [trigger_name for trigger_name in self.get_triggers(model.state)
if any(all(c.check(e) for c in t.conditions)
for ts in self.events[trigger_name].transitions.values()
for t in ts)]
# override Machine.add_model to assign 'can_trigger' to the model
def add_model(self, model, initial=None):
super(PeekMachine, self).add_model(model, initial)
setattr(model, 'can_trigger', partial(self._can_trigger, model))
states = ['A', 'B', 'C', 'D']
transitions = [
dict(trigger='go_A', source='*', dest='A', conditions=['depends_on']), # only available when condition=True is passed
dict(trigger='go_B', source='*', dest='B', conditions=['success']), # always available
dict(trigger='go_C', source='*', dest='C', conditions=['fails']), # never available
dict(trigger='go_D', source='*', dest='D', conditions=['is_state_B']), # only available in state B
dict(trigger='reset', source='D', dest='A', conditions=['success', 'depends_on']), # only available in state D when condition=True is passed
dict(trigger='forwards', source='A', dest='D', conditions=['success', 'fails']), # never available
]
model = Model()
machine = PeekMachine(model, states=states, transitions=transitions, initial='A', auto_transitions=False)
assert model.can_trigger() == ['go_B']
assert set(model.can_trigger(condition=True)) == set(['go_A', 'go_B'])
model.go_B(condition=True)
assert set(model.can_trigger()) == set(['go_B', 'go_D'])
model.go_D()
assert model.can_trigger() == ['go_B']
assert set(model.can_trigger(condition=True)) == set(['go_A', 'go_B', 'reset'])
In [ ]: